package com.changdu.seatunnel.admin.handlers;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import com.changdu.seatunnel.admin.config.SeatunnelQueryProperty;
import com.changdu.seatunnel.admin.config.enums.SyncTaskStatus;
import com.changdu.seatunnel.admin.config.enums.SyncType;
import com.changdu.seatunnel.admin.entity.SyncTask;
import com.changdu.seatunnel.admin.pojo.SeatunnelTaskInfo;
import com.changdu.seatunnel.admin.service.KafkaCreater;
import com.changdu.seatunnel.admin.service.SyncTaskService;
import com.changdu.seatunnel.admin.util.JsonUtils;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

import lombok.extern.slf4j.Slf4j;

/**
 * job 任务状态扫描
 * @author wangxiaoliang
 * 2025年8月28日
 */
@Slf4j
public class JobStateScanHandler implements InitializingBean, DisposableBean {

    private static final ScheduledExecutorService SCHEDULE_EXECUTOR = 
            Executors.newScheduledThreadPool(
                    2, new ThreadFactoryBuilder().setNameFormat("seatunnel-task-listener").build());
    
    private final List<Future<?>> futures = new ArrayList<>();	
	private final SeatunnelQueryProperty config;
	private final SyncTaskService syncTaskService;
	private final DingRobotHandler dingRobotHandler;
	private final KafkaCreater kafkaClient;
	
	public JobStateScanHandler(SeatunnelQueryProperty seatunnelQueryProperty, DingRobotHandler dingRobotHandler,
			SyncTaskService syncTaskService, KafkaCreater kafkaClient) {
		this.config = seatunnelQueryProperty;
		this.dingRobotHandler = dingRobotHandler;
		this.syncTaskService = syncTaskService;
		this.kafkaClient = kafkaClient;
	}
	
	/**
	 * 扫描全量任务并将任务状态进行更新
	 * @throws InterruptedException 
	 */
	public void checkTaskState() throws InterruptedException {
		// 只存在运行状态中的任务，非运行中的任务状态变更通过手动操作完成
		List<SyncTask> syncTasks = syncTaskService.getRunningTasks();
		for (SyncTask syncTask : syncTasks) {
			SeatunnelTaskInfo taskStatus = syncTaskService.syncTaskState(syncTask);
			// 对异常状态发送告警通知
			if (SyncTaskStatus.NOT_EXISTS == taskStatus.taskStatus() || 
					SyncTaskStatus.FAILED == taskStatus.taskStatus()) {
				dingRobotHandler.sendDingTalkMessage(syncTask, "--");
			}
			Long lag = showKafkaLag(syncTask);
			if (lag > config.getScheduled().getMaxlag()) {
				dingRobotHandler.sendDingTalkMessage(syncTask, lag.toString());
			}
			Thread.sleep(Duration.ofSeconds(1));
		}
	}

	/**
	 * 获取 kafka lag
	 * @param syncTask
	 */
	public Long showKafkaLag(SyncTask syncTask) {
		SyncType syncType = SyncType.fromCode(syncTask.getSourcePlugin(), syncTask.getSinkPlugin());
		if (syncType == SyncType.KAFKA_STARROCKS) {
			JsonNode template = JsonUtils.getJsonNode(syncTask.getTemplateContent().getBytes());
			JsonNode source = template.get("source").get(0);
			return kafkaClient.getLagFromTopic(source);
		}
		return 0L;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
    	ScheduledFuture<?> scheduledFuture = SCHEDULE_EXECUTOR.scheduleAtFixedRate(
                () -> {
                    try {
                    	checkTaskState();
                    } catch (Throwable e) {
                        log.error(e.getMessage(), e);
                    }
                },
                60L,
                config.getScheduled().getTaskScan(),
                TimeUnit.SECONDS);
    	futures.add(scheduledFuture);
	}
	
	@Override
	public void destroy() throws Exception {
		futures.forEach(f -> f.cancel(false));
	}	
	
}
